package dyyx.util;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.Set;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.DescribeGroupsRequest;
import org.apache.kafka.common.requests.DescribeGroupsResponse;
import org.apache.kafka.common.requests.ListGroupsRequest;
import org.apache.kafka.common.requests.ListGroupsResponse;
import org.apache.kafka.common.requests.ListGroupsResponse.Group;
import org.apache.kafka.common.requests.OffsetFetchRequest;
import org.apache.kafka.common.requests.OffsetFetchResponse;
import org.apache.kafka.common.requests.OffsetFetchResponse.PartitionData;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.requests.RequestHeader;
import org.apache.kafka.common.requests.ResponseHeader;

import com.alibaba.fastjson.JSON;

import dyyx.dto.PartitionOffsetInfo;


public abstract class KafkaUtil {
	private static final Random RAND = new Random();
	
	
	public static synchronized String genShowOffsetGroupId(){
		
		return "show_offset_group_"+System.currentTimeMillis()+"_"+RAND.nextInt(1000);
	}
	
    public static void close(KafkaConsumer consumer){
    	if(consumer==null){
    		return;
    	}
    	try{
    		consumer.close();
    	}catch(Throwable e){
    		RunStatusUtil.logError("kafkaConsumerCloseError", e);
    	}
		
	}
	
    public static  PartitionOffsetInfo getPartitionOffsetInfo(String servers,Collection<TopicPartition> partitions){
		if(StringUtils.isBlank(servers)){
			throw new RuntimeException("servers blank");
		}
    	if(partitions==null || partitions.isEmpty()){
			// return null;
			throw new RuntimeException("TopicPartitions empty");
		}
        Properties props = new Properties();
		
		String groupId = genShowOffsetGroupId();

		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
		props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
		// props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

		KafkaConsumer<String, String> consumer = null;
		
		try{
		consumer = new KafkaConsumer<String, String>(props);
		consumer.assign(partitions);

		
		Map<TopicPartition, Long> beginningOffsets = consumer.beginningOffsets(partitions);
		Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitions);
		
		PartitionOffsetInfo info = new PartitionOffsetInfo();
		info.beginningOffsets = beginningOffsets;
		info.endOffsets = endOffsets;
		
		return info;
		
		}finally{
			close(consumer);
		}
	}
	

	
	@Deprecated
	public static ByteBuffer send(String host, int port, AbstractRequest request, ApiKeys apiKey) throws IOException {
		Socket socket = connect(host, port);
		try {
			return send(request, apiKey, socket);
		} finally {
			socket.close();
		}
	}

	private static byte[] issueRequestAndWaitForResponse(Socket socket, byte[] request) throws IOException {
		sendRequest(socket, request);
		return getResponse(socket);
	}

	private static void sendRequest(Socket socket, byte[] request) throws IOException {
		DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
		dos.writeInt(request.length);
		dos.write(request);
		dos.flush();
	}

	private static byte[] getResponse(Socket socket) throws IOException {
		DataInputStream dis = null;
		try {
			dis = new DataInputStream(socket.getInputStream());
			byte[] response = new byte[dis.readInt()];
			dis.readFully(response);
			return response;
		} finally {
			if (dis != null) {
				dis.close();
			}
		}
	}

	private static Socket connect(String hostName, int port) throws IOException {
		return new Socket(hostName, port);
	}

	private static ByteBuffer send(AbstractRequest request, ApiKeys apiKey, Socket socket) throws IOException {
		RequestHeader header = new RequestHeader(apiKey, request.version(), "client-id", 0);

		ByteBuffer buffer = request.serialize(header);

		/*
		 * ByteBuffer buffer = ByteBuffer.allocate(header.sizeOf() +
		 * request.sizeOf()); header.writeTo(buffer); request.writeTo(buffer);
		 */

		byte[] serializedRequest = buffer.array();
		byte[] response = issueRequestAndWaitForResponse(socket, serializedRequest);
		ByteBuffer responseBuffer = ByteBuffer.wrap(response);
		ResponseHeader.parse(responseBuffer);
		return responseBuffer;
	}

	public static void main(String[] args) throws Exception {
		
		System.out.println(ApiKeys.LIST_GROUPS.oldestVersion()+","+ApiKeys.LIST_GROUPS.latestVersion());

		short version = (short)1;
		// version = ApiKeys.LIST_GROUPS.latestVersion();
		
		ListGroupsRequest listGroupsRequest = new ListGroupsRequest.Builder().build(version);
		
		ByteBuffer response = send("localhost", 9092, listGroupsRequest, ApiKeys.LIST_GROUPS);
		
		ListGroupsResponse listGroupsResponse = ListGroupsResponse.parse(response, version);
		
		System.out.println(listGroupsResponse);
		
		List<Group> groups = listGroupsResponse.groups();
		for(Group item:groups){
			System.out.println(item.groupId()+","+item.protocolType());
		}
		///
		List<String> groupIds = new ArrayList<>();
		groupIds.add("kafkademo_test_group");
		DescribeGroupsRequest describeGroupsRequest = new DescribeGroupsRequest.Builder(groupIds).build(version);
		response = send("localhost", 9092, describeGroupsRequest, ApiKeys.DESCRIBE_GROUPS);
		
		DescribeGroupsResponse describeGroupsResponse = DescribeGroupsResponse.parse(response, version);
		
		System.out.println(describeGroupsResponse.groups());

		System.out.println(JSON.toJSONString(describeGroupsResponse));

		
		// ListOffsetRequest listOffsetRequest = new ListOffsetRequest()
		
		// Exception in thread "main" org.apache.kafka.common.errors.UnsupportedVersionException: The broker only supports OffsetFetchRequest v1, but we need v2 or newer to request all topic partitions.

		String groupId = "kafkademo_test_group";
		OffsetFetchRequest offsetFetchRequest = OffsetFetchRequest.Builder.allTopicPartitions(groupId).build(version);
		response = send("localhost", 9092, offsetFetchRequest, ApiKeys.OFFSET_FETCH);
		
		OffsetFetchResponse offsetFetchResponse = OffsetFetchResponse.parse(response, version);
		System.out.println(offsetFetchResponse);

		Map<TopicPartition, PartitionData> responseData = offsetFetchResponse.responseData();
		
		System.out.println(responseData);
		
		Set<TopicPartition> keys = responseData.keySet();
		for(TopicPartition item:keys){
			PartitionData pd = responseData.get(item);
			System.out.println(item.topic()+"-"+item.partition()+"="+pd.offset+","+pd.metadata);

		}

	}

}
